#include "message_service_handler.h"
#include "message_db_wrapper.h"
#include "base/logging.h"
#include "base/format_time.h"
#include "base/json_escape.h"
#include <boost/algorithm/string.hpp>
#include "im/adapters/sock_message_outlet_adapter.h"
#include "im/adapters/group_adapter.h"
//#include "im/adapters/comet_message_outlet_adapter.h"
#include "message_distribute_task.h"
#include "ImServiceCommon_constants.h"
#include "im/adapters/UserInfoAdapter.h"

#include <sstream>

namespace imserver {

int64_t MessageServiceHandler::insertMsgDB(ImMessage& message) {
  int64_t msg_id = MessageIdCreator::instance().createMsgId();
  if (msg_id < 0) {
    LOG_ERROR("MessageServiceHandler::insertMsgDB --> create msg_id failed.");
    return -1L;
  }
  message.id = msg_id;

  int64_t dialogue_id = 0;
  if (message.type == ImMessageType::IM_MESSAGE_P2P ||
      message.type == ImMessageType::IM_MESSAGE_RESUME ||
      message.type == ImMessageType::IM_MESSAGE_RESUME_LIKE ||
      message.type == ImMessageType::IM_MESSAGE_RESUME_IGNORE ||
      message.type == ImMessageType::IM_MESSAGE_JOB_INVITATION) {
    if (message.dialogue_id <= 0) {
      if (!MessageDBWrapper::instance().select_tb_dialogue(message.sender, message.receiver, dialogue_id)) {
        return -1L;
      }
      if (dialogue_id <= 0) {
        dialogue_id = MessageIdCreator::instance().createDialogueId();
        if (dialogue_id <= 0) {
          return -1L;
        }
      }
      message.dialogue_id = dialogue_id;
    }
    MessageDBWrapper::instance().update_tb_dialogue(message.dialogue_id, message.sender, message.receiver, msg_id);
    if (!MessageDBWrapper::instance().insert_tb_message_entity(message)) {
      return -1L;
    }
    MessageDBWrapper::instance().insert_tb_p2p_message_index(message);
  } else if (message.type == ImMessageType::IM_MESSAGE_GROUP ||
             message.type == ImMessageType::IM_MESSAGE_GROUP_JOIN ||
             message.type == ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE ||
             message.type == ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION) {
    if (message.group_id <= 0) {
      LOG_ERROR("MessageServiceHandler::insertMsgDB --> msg type is IM_MESSAGE_GROUP, but group_id <= 0. msg_id:" <<
          message.id);
      return -1L;
    }
    if (!MessageDBWrapper::instance().insert_tb_message_entity(message)) {
      return -1L;
    }
    MessageDBWrapper::instance().insert_tb_group_message_index(message);
    if (message.type != ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE) {
      MessageDBWrapper::instance().update_tb_group_message_view_record(message.group_id, message.sender, message.id);
    }
  }
  LOG_INFO("MessageServiceHandler::insertMsgDB --> ok. msg_id:" << message.id << " type:" << message.type << " dialogue_id:"
      << message.dialogue_id);

  return msg_id;
}

bool MessageServiceHandler::insertRedis(const ImMessage& message) {
  /*
  std::ostringstream os;
  if (message.type == ImMessageType::IM_MESSAGE_P2P || message.type == ImMessageType::IM_SYSTEM_NOTIFY) {
    os << "ZADD " <<
          "from:" << message.sender << ":to:" << message.receiver << " " <<
          message.id << " " <<
          "{\"id\":\"" << message.id << "\",\"type\":\"" << message.type << "\"," <<
          "\"flags\":" << message.flags << "\"}";
  } else if (message.type == ImMessageType::IM_MESSAGE_GROUP) {
    os << "ZADD " <<
          "group_" << message.group_id << " " <<
          message.id << " " <<
          "{\"id\":\"" << message.id << "\",\"type\":\"" << message.type << "\"," <<
          "\"flags\":" << message.flags << "\"," <<
          "\"sender\":\"" << message.sender << "\"}";
  } else if (message.type == ImMessageType::IM_SYSTEM_NOTIFY_GLOB) {
    os << "ZADD " <<
          "system_" << message.sender << " " <<
          message.id << " " <<
          "{\"id\":\"" << message.id << "\",\"type\":\"" << message.type << "\"," <<
          "\"flags\":" << message.flags << "\"," <<
          "\"sender\":\"" << message.sender << "\"}";
  } else {
    LOG_ERROR("MessageServiceHandler::insertRedis --> wrong type. msg_id:" << message.id << " msg_type:" << message.type);
    return false;
  }
  redisReply * reply = NULL;
  bool redisRes = false;
  for (int i = 0; i < 2; ++i) {
    reply = redisExecutor.Execute("redis server_key", os.str().c_str());
    if (reply != NULL) {
      if (reply->type == REDIS_REPLY_INTEGER) {
        redisRes = true;
      }
      freeReplyObject(reply);
      if (redisRes) {
        break;
      }
    }
  }
  if (!redisRes) {
    LOG_ERROR("MessageServiceHandler::insertRedis --> redis zadd failed. msg_id:" << message.id);
    return false;
  }

  redisRes = false;
  os.str("");
  if (message.type == ImMessageType::IM_MESSAGE_P2P || ImMessageType::IM_SYSTEM_NOTIFY) {
    os << "ZCARD from:" << message.sender << ":to:" << message.receiver;
  } else if (message.type == ImMessageType::IM_MESSAGE_GROUP) {
    os << "ZCARD group_" << message.group_id;
  } else if (message.type == ImMessageType::IM_SYSTEM_NOTIFY_GLOB) {
    os << "ZCARD system_" << message.sender;
  }
  int cur_msg_num = 0;
  for (int i = 0; i < 2; ++i) {
    reply = redisExecutor.Execute("redis server_key", os.str().c_str());
    if (reply != NULL) {
      if (reply->type == REDIS_REPLY_INTEGER) {
        cur_msg_num = reply->integer;
        redisRes = true;

      }
      freeReplyObject(reply);
      if (redisRes) {
        break;
      }
    }
  }
  if (!redisRes) {
    LOG_ERROR("MessageServiceHandler::insertRedis --> redis zcard failed. msg_id:" << message.id);
  }
  os.str("");
  redisRes = false;
  if (cur_msg_num > 1000) {
    if (message.type == ImMessageType::IM_MESSAGE_P2P || message.type == ImMessageType::IM_SYSTEM_NOTIFY) {
      os << "ZREMRANGEBYRANK from:" << message.sender << ":to:" << message.receiver << " 0 0";
    } else if (message.type == ImMessageType::IM_MESSAGE_GROUP) {
      os << "ZREMRANGEBYRANK group_" << message.group_id << " 0 0";
    } else if (message.type == ImMessageType::IM_SYSTEM_NOTIFY_GLOB) {
      os << "ZREMRANGEBYRANK system_" << message.sender << " 0 0";
    }
    for (int i = 0; i < 2; ++i) {
      reply = redisExecutor.Execute("redis server_key", os.str().c_str());
      if (reply != NULL) {
        if (reply->type == REDIS_REPLY_INTEGER) {
          redisRes = true;
        }
        freeReplyObject(reply);
        if (redisRes) {
          break;
        }
      }
    }
    if (!redisRes) {
      LOG_ERROR("MessageServiceHandler::insertRedis --> redis zremrangebyrank failed. msg_id:" << message.id);
    }
  }

  os.str("");
  redisRes = false;
  if (message.type == ImMessageType::IM_MESSAGE_P2P || message.type == ImMessageType::IM_MESSAGE_GROUP) {
    os << "ZADD contact_list:" << message.receiver << " " << message.id << " ";
    if (message.type == ImMessageType::IM_MESSAGE_P2P) {
      os << message.sender;
    } else if (message.type == ImMessageType::IM_MESSAGE_GROUP) {
      os << "group_"<< message.group_id;
    }
    for (int i = 0; i < 2; ++i) {
      reply = redisExecutor.Execute("redis server_key", os.str().c_str());
      if (reply != NULL) {
        if (reply->type == REDIS_REPLY_INTEGER) {
          redisRes = true;
        }
        freeReplyObject(reply);
        if (redisRes) {
          break;
        }
      }
    }
    if (!redisRes) {
      LOG_ERROR("MessageServiceHandler::insertRedis --> redis contact_list zadd failed. msg_id:" << message.id);
    }
  }
  */

  return true;
}

void MessageServiceHandler::CreateMessage(std::map<int64_t, int64_t>& ret_, const ImMessage& message) {
  LOG_INFO("MessageServiceHandler::CreateMessage --> try to create msg. type:" << message.type << " from:" <<
      message.sender << " to:" << message.receiver << " group_id:" << message.group_id);
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  using apache::thrift::concurrency::Runnable;
  if (message.type == ImMessageType::IM_MESSAGE_TYPING) {
    base::GetThreadManager()->add(boost::shared_ptr<Runnable>(new MessageDistributeTask(message)));
    return;
  }
  std::map<int64_t, int64_t> res;
  ImMessage msg(message);
  if (msg.type == ImMessageType::IM_MESSAGE_GROUP ||
      msg.type == ImMessageType::IM_MESSAGE_GROUP_JOIN ||
      msg.type == ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE ||
      msg.type == ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION) {
    msg.dialogue_id = msg.group_id;
  }
  msg.timestamp = (int64_t)time(0);
  int64_t msg_id = insertMsgDB(msg);
  if (msg_id <= 0) {
    return;
  }
  insertRedis(msg);
  ret_.insert(std::make_pair(msg_id, msg.dialogue_id));

  base::GetThreadManager()->add(boost::shared_ptr<Runnable>(new MessageDistributeTask(msg)));
  LOG_INFO("MessageServiceHandler::CreateMessage --> create msg ok. type:" << message.type << " from:" <<
      message.sender << " to:" << message.receiver << " dialogue_id:" << message.group_id);
}

void MessageServiceHandler::getUnreadDialogueMessageInfo(int64_t uid,
    std::map<int64_t, std::pair<int32_t, int64_t> >& res) {
  std::set<int64_t> dialogue_ids;
  if (MessageDBWrapper::instance().select_tb_dialogue(uid, -1L, dialogue_ids)) {
    MessageDBWrapper::instance().select_tb_p2p_message_index_u(uid, dialogue_ids,
        ImMessageFlag::IM_MESSAGE_READ, res);
  }
}

void MessageServiceHandler::getUnreadGroupMessageInfo(int64_t uid,
    std::map<int64_t, std::pair<int32_t, int64_t> >& res) {
  std::set<int64_t> group_ids;
  GroupAdapter::instance().GetJoinedGroupIds(group_ids, uid);
  for (std::set<int64_t>::const_iterator it = group_ids.begin(); it != group_ids.end(); ++it) {
    int64_t last_read_msgid = 0;
    if (!MessageDBWrapper::instance().select_tb_group_message_view_record(*it, uid, last_read_msgid)) {
      continue;
    }
    int unread_count = 0;
    int64_t max_msgid = 0;
    if (!MessageDBWrapper::instance().select_tb_group_message_index_u(uid, *it, last_read_msgid,
        unread_count, max_msgid)) {
      continue;
    }
    if (unread_count <= 0) {
      continue;
    }
    res.insert(std::make_pair(*it, std::make_pair(unread_count, max_msgid)));
  }
}

void MessageServiceHandler::GetUnReadMessageInfo(std::string& ret_, int64_t uid) {
  LOG_INFO("MessageServiceHandler::GetAllUnReadMessageInfo --> uid:" << uid);
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  std::map<int64_t, std::pair<int32_t, int64_t> > dialogue_res;
  getUnreadDialogueMessageInfo(uid, dialogue_res);

  std::map<int64_t, std::pair<int32_t, int64_t> > group_res;
  getUnreadGroupMessageInfo(uid, group_res);

  std::map<int64_t, int64_t> msgid_dialogueid;
  BOOST_AUTO(it1, dialogue_res.begin());
  for (; it1 != dialogue_res.end(); ++it1) {
    msgid_dialogueid.insert(std::make_pair((it1->second).second, it1->first));
  }
  BOOST_AUTO(it2, group_res.begin());
  for (; it2 != group_res.end(); ++it2) {
    msgid_dialogueid.insert(std::make_pair((it2->second).second, it2->first));
  }
  std::map<int64_t, ImMessage> msg_res;
  MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_dialogueid, msg_res);

  std::ostringstream os;
  os << "[";
  BOOST_AUTO(it3, msg_res.begin());
  for (; it3 != msg_res.end(); ++it3) {
    if (it3 != msg_res.begin()) {
      os << ",";
    }
    user::UserBaseInfo userInfo;
    user::UserAdapter::Instance().getUserBasicInfo(userInfo, (it3->second).sender);
    std::string from  = boost::lexical_cast<std::string>((it3->second).sender);
    std::string fromname;
    std::string fromimg;
    if ((it3->second).group_id > 0) {
      from += ("@" + boost::lexical_cast<std::string>((it3->second).group_id));
      GroupInfo group_info;
      GroupAdapter::instance().GetGroupInfo(group_info, (it3->second).group_id);
      fromname = group_info.name;
      fromimg = group_info.logo;
    } else {
      user::UserBaseInfo userInfo;
      user::UserAdapter::Instance().getUserBasicInfo(userInfo, (it3->second).sender);
      fromname = userInfo.nick_name;
      fromimg = userInfo.photo_url;
    }
    std::string body = base::JsonEscape((it3->second).body);
    if ((it3->second).type == ImMessageType::IM_MESSAGE_RESUME) {
      body = g_ImServiceCommon_constants.IM_MESSAGE_RESUME_SUMMARY;
    } else if ((it3->second).type == ImMessageType::IM_MESSAGE_JOB_INVITATION ||
               (it3->second).type == ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION) {
      body = g_ImServiceCommon_constants.IM_MESSAGE_JOB_INVITATION_SUMMARY;
    }

    os << //"{\"max_mid\":" << (it3->second).id << "," <<
        "{\"max_mbody\":\"" << body << "\"," <<
        "\"max_mtime\":" << (it3->second).timestamp << "," <<
        "\"from\":\"" << from << "\"," <<
        "\"fromname\":\"" << fromname << "\"," <<
        "\"fromimg\":\"" << fromimg << "\",";
    if ((it3->second).group_id > 0) {
      os << "\"unread_count\":" << group_res[(it3->second).group_id].first << ",";
    } else {
      os << "\"unread_count\":" << dialogue_res[(it3->second).dialogue_id].first << ",";
    }
    os <<"\"did\":" << (it3->second).dialogue_id << "}";
  }
  os << "]";
  ret_ = os.str();
  LOG_INFO("MessageServiceHandler::GetAllUnReadMessageInfo --> uid:" << uid << " res_count:" << msg_res.size());
  LOG_DEBUG("MessageServiceHandler::GetAllUnReadMessageInfo --> uid:" << uid << " ret_:" << ret_);
}

void MessageServiceHandler::GetP2PHistoryMessage(std::string& ret_, int64_t uid, int64_t another_uid, int64_t min_msg_id, int32_t limit) {
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  int64_t dialogue_id = 0;
  if (!MessageDBWrapper::instance().select_tb_dialogue(uid, another_uid, dialogue_id)) {
    ret_ = "[]";
    return;
  }
  std::set<int64_t> msgid_res;
  if (!MessageDBWrapper::instance().select_tb_p2p_message_index_o(uid, dialogue_id, min_msg_id, limit, msgid_res)) {
    ret_ = "[]";
    return;
  }
  std::map<int64_t, int64_t> msgid_dialogueid;
  for (std::set<int64_t>::const_iterator it = msgid_res.begin(); it != msgid_res.end(); ++it) {
    msgid_dialogueid.insert(std::make_pair(*it, dialogue_id));
  }
  std::map<int64_t, ImMessage> msg_res;
  if (!MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_dialogueid, msg_res)) {
    ret_ = "[]";
    return;
  }

  std::string fromname;
  std::string fromimg;

  std::ostringstream os;
  os << "[";
  BOOST_AUTO(it, msg_res.begin());
  for (; it != msg_res.end(); ++it) {
    user::UserBaseInfo userInfo;
    user::UserAdapter::Instance().getUserBasicInfo(userInfo, (it->second).sender);
    fromname =  userInfo.nick_name;
    fromimg = userInfo.photo_url;

    if (it != msg_res.begin()) {
      os << ",";
    }
    os << "{\"mid\":" << it->first << "," <<
        "\"type\":" << (it->second).type << "," <<
        "\"body\":\"" << base::JsonEscape((it->second).body) << "\"," <<
        "\"from\":" << (it->second).sender << "," <<
        "\"fromname\":\"" << fromname << "\"," <<
        "\"fromimg\":\"" << fromimg << "\"," <<
        "\"to\":" << (it->second).receiver << "," <<
        "\"time\":" << (it->second).timestamp << "," <<
        "\"did\":" << (it->second).dialogue_id << "}";
  }
  os << "]";

  ret_ = os.str();
  LOG_INFO("MessageServiceHandler::GetP2PHistoryMessage --> uid:" << uid << " another_uid:" << another_uid
      << " min_msg_id:" << min_msg_id << " limit:" << limit << " res_count:" << msg_res.size());
  LOG_INFO("MessageServiceHandler::GetP2PHistoryMessage --> uid:" << uid << " another_uid:" << another_uid
      << " min_msg_id:" << min_msg_id << " limit:" << limit << " ret_:" << ret_);
}

void MessageServiceHandler::GetP2PNewMessage(std::string& ret_, int64_t uid, int64_t another_uid, int64_t max_msg_id, int32_t limit) {
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  int64_t dialogue_id = 0;
  if (!MessageDBWrapper::instance().select_tb_dialogue(uid, another_uid, dialogue_id)) {
    ret_ = "[]";
    return;
  }
  std::set<int64_t> msgid_res;
  if (!MessageDBWrapper::instance().select_tb_p2p_message_index_n(uid, dialogue_id, max_msg_id, limit, msgid_res)) {
    ret_ = "[]";
    return;
  }
  std::map<int64_t, int64_t> msgid_dialogueid;
  for (std::set<int64_t>::const_iterator it = msgid_res.begin(); it != msgid_res.end(); ++it) {
    msgid_dialogueid.insert(std::make_pair(*it, dialogue_id));
  }
  std::map<int64_t, ImMessage> msg_res;
  if (!MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_dialogueid, msg_res)) {
    ret_ = "[]";
    return;
  }
  if (!msg_res.empty()) {
    MessageDBWrapper::instance().update_tb_p2p_message_index(uid, dialogue_id, (msg_res.rbegin())->first, ImMessageFlag::IM_MESSAGE_READ);
  } else {
    MessageDBWrapper::instance().update_tb_p2p_message_index(uid, dialogue_id, max_msg_id, ImMessageFlag::IM_MESSAGE_READ);
  }
  std::string fromname;
  std::string fromimg;

  std::ostringstream os;
  os << "[";
  BOOST_AUTO(it, msg_res.begin());
  for (; it != msg_res.end(); ++it) {
    user::UserBaseInfo userInfo;
    user::UserAdapter::Instance().getUserBasicInfo(userInfo, (it->second).sender);
    fromname =  userInfo.nick_name;
    fromimg = userInfo.photo_url;

    if (it != msg_res.begin()) {
      os << ",";
    }
    os << "{\"mid\":" << it->first << "," <<
        "\"type\":" << (it->second).type << "," <<
        "\"body\":\"" << base::JsonEscape((it->second).body) << "\"," <<
        "\"from\":" << (it->second).sender << "," <<
        "\"fromname\":\"" << fromname << "\"," <<
        "\"fromimg\":\"" << fromimg << "\"," <<
        "\"to\":" << (it->second).receiver << "," <<
        "\"time\":" << (it->second).timestamp << "," <<
        "\"did\":" << (it->second).dialogue_id << "}";
  }
  os << "]";

  ret_ = os.str();
  LOG_INFO("MessageServiceHandler::GetP2PNewMessage --> uid:" << uid << " another_uid:" << another_uid
      << " max_msg_id:" << max_msg_id << " limit:" << limit << " res_count:" << msg_res.size());
  LOG_INFO("MessageServiceHandler::GetP2PNewMessage --> uid:" << uid << " another_uid:" << another_uid
      << " max_msg_id:" << max_msg_id << " limit:" << limit << " ret_:" << ret_);
}

void MessageServiceHandler::GetGroupHistoryMessage(std::string& ret_, int64_t uid, int64_t group_id, int64_t min_msg_id, int32_t limit) {
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  std::set<int64_t> msgid_res;
  if (!MessageDBWrapper::instance().select_tb_group_message_index_o(uid, group_id, min_msg_id, limit, msgid_res)) {
    ret_ = "[]";
    return;
  }
  std::map<int64_t, int64_t> msgid_groupid;
  for (std::set<int64_t>::const_iterator it = msgid_res.begin(); it != msgid_res.end(); ++it) {
    msgid_groupid.insert(std::make_pair(*it, group_id));
  }

  std::map<int64_t, ImMessage> msg_res;
  if (!MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_groupid, msg_res)) {
    ret_ = "[]";
    return;
  }
  if (msg_res.empty()) {
    ret_ = "[]";
    return;
  }

  std::string fromname;
  std::string fromimg;

  std::ostringstream os;
  os << "[";
  BOOST_AUTO(it, msg_res.begin());
  for (; it != msg_res.end(); ++it) {
    user::UserBaseInfo userInfo;
    user::UserAdapter::Instance().getUserBasicInfo(userInfo, (it->second).sender);
    fromname =  userInfo.nick_name;
    fromimg = userInfo.photo_url;

    if (it != msg_res.begin()) {
      os << ",";
    }
    std::string from = boost::lexical_cast<std::string>((it->second).sender) + "@"
        + boost::lexical_cast<std::string>((it->second).group_id);
    os << "{\"msg_id\":" << it->first << "," <<
        "\"type\":" << (it->second).type << "," <<
        "\"body\":\"" << base::JsonEscape((it->second).body) << "\"," <<
        "\"from\":" << (it->second).sender << "," <<
        "\"fromname\":\"" << fromname << "\"," <<
        "\"fromimg\":\"" << fromimg << "\"," <<
        "\"timestamp\":" << (it->second).timestamp << "," <<
        "\"group_id\":" << (it->second).group_id << "}";
  }
  os << "]";
  ret_ = os.str();
  LOG_INFO("MessageServiceHandler::GetGroupHistoryMessage --> uid:" << uid << " group_id:" << group_id
      << " min_msg_id:" << min_msg_id << " limit:" << limit << " res_count:" << msg_res.size());
  LOG_INFO("MessageServiceHandler::GetGroupNewMessage --> uid:" << uid << " group_id:" << group_id
      << " min_msg_id:" << min_msg_id << " limit:" << limit << " ret_:" << ret_);
}


void MessageServiceHandler::GetGroupNewMessage(std::string& ret_, int64_t uid, int64_t group_id, int64_t max_msg_id, int32_t limit) {
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  std::set<int64_t> msgid_res;
  if (!MessageDBWrapper::instance().select_tb_group_message_index_n(uid, group_id, max_msg_id, limit, msgid_res)) {
    ret_ = "[]";
    return;
  }
  std::map<int64_t, int64_t> msgid_groupid;
  for (std::set<int64_t>::const_iterator it = msgid_res.begin(); it != msgid_res.end(); ++it) {
    msgid_groupid.insert(std::make_pair(*it, group_id));
  }

  std::map<int64_t, ImMessage> msg_res;
  if (!MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_groupid, msg_res)) {
    ret_ = "[]";
    return;
  }
  if (msg_res.empty()) {
    ret_ = "[]";
    MessageDBWrapper::instance().update_tb_group_message_view_record(group_id, uid, max_msg_id);
    return;
  }

  MessageDBWrapper::instance().update_tb_group_message_view_record(group_id, uid, (msg_res.rbegin())->first);

  std::string fromname;
  std::string fromimg;

  std::ostringstream os;
  os << "[";
  BOOST_AUTO(it, msg_res.begin());
  for (; it != msg_res.end(); ++it) {
    user::UserBaseInfo userInfo;
    user::UserAdapter::Instance().getUserBasicInfo(userInfo, (it->second).sender);
    fromname =  userInfo.nick_name;
    fromimg = userInfo.photo_url;

    if (it != msg_res.begin()) {
      os << ",";
    }
    std::string from = boost::lexical_cast<std::string>((it->second).sender) + "@"
        + boost::lexical_cast<std::string>((it->second).group_id);
    os << "{\"mid\":" << it->first << "," <<
        "\"type\":" << (it->second).type << "," <<
        "\"body\":\"" << base::JsonEscape((it->second).body) << "\"," <<
        "\"from\":" << (it->second).sender << "," <<
        "\"fromname\":\"" << fromname << "\"," <<
        "\"fromimg\":\"" << fromimg << "\"," <<
        "\"time\":" << (it->second).timestamp << "," <<
        "\"did\":" << (it->second).group_id << "}";
  }
  os << "]";
  ret_ = os.str();
  LOG_INFO("MessageServiceHandler::GetGroupHistoryMessage --> uid:" << uid << " group_id:" << group_id
      << " max_msg_id:" << max_msg_id << " limit:" << limit << " res_count:" << msg_res.size());
  LOG_INFO("MessageServiceHandler::GetGroupNewMessage --> uid:" << uid << " group_id:" << group_id
      << " max_msg_id:" << max_msg_id << " limit:" << limit << " ret_:" << ret_);
}

void MessageServiceHandler::SetMessageReceiveStatus(int64_t uid, int64_t msg_id, int64_t dialogue_id) {
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  std::map<int64_t, int64_t> msgid_dialogid;
  msgid_dialogid.insert(std::make_pair(msg_id, dialogue_id));
  std::map<int64_t, ImMessage> msg_res;
  if (!MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_dialogid, msg_res) || msg_res.empty()) {
    return;
  }

  if (msg_res[msg_id].type == ImMessageType::IM_MESSAGE_P2P ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_RESUME ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_RESUME_LIKE ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_RESUME_IGNORE ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_JOB_INVITATION) {
    MessageDBWrapper::instance().update_tb_p2p_message_index(msg_res[msg_id].receiver, msg_res[msg_id].dialogue_id,
        msg_id, ImMessageFlag::IM_MESSAGE_RECEIVED);
  }
  LOG_INFO("MessageServiceHandler::SetMessageReceiveStatus --> uid:" << uid << " msg_id:" << msg_id << 
      " dialogue_id:" << dialogue_id);
}

void MessageServiceHandler::SetMessageReadStatus(int64_t uid, int64_t msg_id, int64_t dialogue_id) {
  base::SimpleMysqlThreadInvoker mysqlInvoker;
  std::map<int64_t, int64_t> msgid_dialogid;
  msgid_dialogid.insert(std::make_pair(msg_id, dialogue_id));
  std::map<int64_t, ImMessage> msg_res;
  if (!MessageDBWrapper::instance().select_tb_message_entity(uid, msgid_dialogid, msg_res) || msg_res.empty()) {
    return;
  }

  if (msg_res[msg_id].type == ImMessageType::IM_MESSAGE_P2P ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_RESUME ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_RESUME_LIKE ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_RESUME_IGNORE ||
      msg_res[msg_id].type == ImMessageType::IM_MESSAGE_JOB_INVITATION) {
    MessageDBWrapper::instance().update_tb_p2p_message_index(msg_res[msg_id].receiver, msg_res[msg_id].dialogue_id,
        msg_id, ImMessageFlag::IM_MESSAGE_READ);
  } else if (msg_res[msg_id].type == ImMessageType::IM_MESSAGE_GROUP ||
             msg_res[msg_id].type == ImMessageType::IM_MESSAGE_GROUP_JOIN ||
             msg_res[msg_id].type == ImMessageType::IM_MESSAGE_GROUP_RESUME_LIKE ||
             msg_res[msg_id].type == ImMessageType::IM_MESSAGE_GROUP_JOB_INVITATION) {
    MessageDBWrapper::instance().update_tb_group_message_view_record(msg_res[msg_id].group_id, uid, msg_id);
  }

  LOG_INFO("MessageServiceHandler::SetMessageReadStatus --> uid:" << uid << " msg_id:" << msg_id << 
      " dialogue_id:" << dialogue_id);
}

}
